package com.ikas.ai.server.kafka;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONValidator;
import com.ikas.ai.model.KafkaMeteData;
import com.ikas.ai.server.module.alarm.service.TripAlarmService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 跳机测点消费存储到t_data表
 */
@Slf4j
@Component
public class KafkaTripMeteConsumerService {

    @Autowired
    private TripAlarmService tripAlarmService;

    // 消费监听
    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_MACHINE_TRIP}, topicPattern = "0", groupId = KafkaInitialConfiguration.TOPIC_MACHINE_TRIP_GROUP)
    public void onMessageMachineTrip(List<ConsumerRecord<String, String>> consumerRecord, Acknowledgment ack) {
        consumerRecord.forEach(record -> {
            try {
                List<KafkaMeteData> meteDataList = new ArrayList<>();
                final JSONValidator.Type type = JSONValidator.from(record.value()).getType();
                if (type == JSONValidator.Type.Array) {
                    meteDataList = JSONArray.parseArray(record.value(), KafkaMeteData.class);
                } else if (type == JSONValidator.Type.Object) {
                    meteDataList = Collections.singletonList(JSONObject.parseObject(record.value(), KafkaMeteData.class));
                }
                if (CollectionUtil.isEmpty(meteDataList)) {
                    return;
                }
                log.info("====TRIP-KAFKA-====>offset>[{}]--timestamp>[{}]------>meteDataList:{}", record.offset()
                        , record.timestamp(), JSONUtil.toJsonStr(meteDataList));
                meteDataList.stream().forEach(mete -> {
                    tripAlarmService.processTripMete(mete);
                });
            } catch (Exception e) {
                log.error("=====跳机信号数据处理错误======", e);
            }
        });
        //手动提交
        ack.acknowledge();
    }
}



















